package sc.analysis.movingavg;


import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import sc.writable.IntermediateData;
import sc.writable.IntermediateData.Type;


/**
 * Reducer of the Moving average analysis. Uses the information about the moving
 * average and the last price to generate alerts and scores.
 * 
 * FIXME moving averages crosses are not well defined (it generates 2 alerts...)
 *
 * @author Quentin Rossé
 *
 */
public  class MaReducer extends Reducer<Text, MaData, 
                                        Text, IntermediateData> {
	
	/**
	 * data written for the next phase
	 */
	private static IntermediateData imd = new IntermediateData();
	
	/**
	 * to store the values to use them more than once
	 */
	private static ArrayList<MaData>  maData = new ArrayList<MaData>();
	
	@Override
	public void reduce(Text key, Iterable<MaData> values, Context context) throws IOException, InterruptedException {
		
		double percentageWithLast;
		
		maData.clear();
		for(MaData ma : values)
			maData.add(new MaData(ma));
			
		for(MaData ma : maData){
			//MA general informations
			percentageWithLast = (ma.getLastPrice() - ma.getAvg()) / ma.getAvg();
			imd.setAnalysisName("SMA"+ma.getNum());
			imd.setScore(score(percentageWithLast));
			imd.setType(ma.getNum());
			imd.setDetails(ma.getLastPrice() + " is " + percentageWithLast * 100 + 
	                 "% from the MA ("+ ma.getAvg()+ ")");
			context.write(key, imd);
			
			//price/MA cross alert
			if(ma.getPrevPrice() <= ma.getPrevAvg() && ma.getLastPrice() > ma.getAvg()){
				imd.setType(Type.ALERT_BULL);
				imd.setScore(1);
				imd.setDetails("Price crossed MA" + ma.getNum());
				context.write(key, imd);
			}else if(ma.getPrevPrice() >= ma.getPrevAvg() && ma.getLastPrice() < ma.getAvg()){
				imd.setType(Type.ALERT_BEAR);
				imd.setScore(1);
				imd.setDetails("Price crossed MA" + ma.getNum());
				context.write(key, imd);
			}
			
			//MA/MA cross alert
			for(MaData ma2 : maData){
				if(ma.getNum() != ma2.getNum()){
					if(ma.getPrevAvg() <= ma2.getPrevAvg() && ma.getAvg() > ma2.getAvg()){
						imd.setType(Type.ALERT_BULL);
						imd.setScore(1);
						imd.setDetails("MA"+ma.getNum()+" crossed MA" + ma2.getNum());
						context.write(key, imd);
					}
					else if(ma.getPrevAvg() >= ma2.getPrevAvg() && ma.getAvg() < ma2.getAvg()){
						imd.setType(Type.ALERT_BEAR);
						imd.setScore(1);
						imd.setDetails("MA"+ma.getNum()+" crossed MA" + ma2.getNum());
						context.write(key, imd);
					}
				}
			}
		}
	}
	
	@Override
	public void setup(Context context){}
	
	@Override
	public void cleanup(Context context){}
	
	/**
	 * Return the score from the percentage between the MA and the last price
	 * 
	 * @param percentageWithLast percentage between the MA and the last price
	 * @return
	 */
	private int score(double percentageWithLast){
		if(percentageWithLast < -0.3)
			return 3;
		else if(percentageWithLast < -0.1)
			return 0;
		else if(percentageWithLast < -0.02)
			return 3;
		else if(percentageWithLast < 0.02)
			return 5;
		else if(percentageWithLast < 0.1)
			return 10;
		else if(percentageWithLast < 0.3)
			return 7;
		else
			return 3;
	}
}